package com.my.kafka.serer;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.common.metrics.stats.Count;
import org.apache.kafka.common.protocol.types.Field.Str;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

/**
 * 生产者
 * @author my
 *
 */
@Component
public class DataProducer {
	
	@Autowired(required = false)
	private KafkaTemplate<String, String> template;
	
	@Value("${kafka_topic}")
	String topic;
	private int Count = 9;
	/**
	 * 发送测试数据
	 * 指定key
	 * 每5秒发送一次
	 */
	@Scheduled(cron = "0/5 * * * * ?")
	public void sendTestMessage() {
		Map<String, Object> testValue = new HashMap<>();
		testValue.put("name", "ruyi-test1");
		template.send(topic, "test-key", JSON.toJSONString(testValue));
	}
	
	/**
	 * 发送测试数据
	 * 不指定key
	 */
	@Scheduled(cron = "0/5 * * * * ?")
	public void sendTestMessage2() {
		Count++;
		Map<String, Object> testValue = new HashMap<>();
		testValue.put("name", "ruyi-test2");
		testValue.put("age", Count);
		System.out.println("生产者："+JSON.toJSONString(testValue));
		send(topic, "test-key2", JSON.toJSONString(testValue));
	}
	
	/**
	 * 发送数据
	 * @param topic
	 * @param key
	 * @param value
	 */
	private void send(String topic, String key, String value) {
		Map<String, String> map = new HashMap<>();
        map.put("key", key);
        map.put("value", value);
        template.send(topic, JSON.toJSONString(map));
	}
}
